/*
 * Copyright (c) 2012-2013 Open Source Community - <http://www.peerfact.org>
 * Copyright (c) 2011-2012 University of Paderborn - UPB
 * Copyright (c) 2005-2011 KOM - Multimedia Communications Lab
 *
 * This file is part of PeerfactSim.KOM.
 * 
 * PeerfactSim.KOM is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * any later version.
 * 
 * PeerfactSim.KOM is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with PeerfactSim.KOM.  If not, see <http://www.gnu.org/licenses/>.
 *
 */

package org.peerfact.impl.overlay.dht.pastry.parallel.components;

import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.Set;

import org.peerfact.api.common.Message;
import org.peerfact.api.overlay.dht.DHTObject;
import org.peerfact.impl.overlay.dht.pastry.base.components.PastryConstants;
import org.peerfact.impl.overlay.dht.pastry.base.components.PastryContact;
import org.peerfact.impl.overlay.dht.pastry.base.components.PastryKey;
import org.peerfact.impl.overlay.dht.pastry.base.components.PastryRoutingTable;
import org.peerfact.impl.overlay.dht.pastry.base.components.TransmissionCallback.Failed;
import org.peerfact.impl.overlay.dht.pastry.base.messages.AckMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.ConfirmStoreMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.FinalJoinAnswerMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.JoinMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.LookupMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.LookupReplyMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.MsgTransInfo;
import org.peerfact.impl.overlay.dht.pastry.base.messages.PastryBaseMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.RequestLeafSetMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.RequestNeighborhooodSetMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.RequestRouteSetMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.StateUpdateMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.StoreMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.ValueLookupMsg;
import org.peerfact.impl.overlay.dht.pastry.base.messages.ValueLookupReplyMsg;
import org.peerfact.impl.overlay.dht.pastry.base.nodestate.RouteSet;
import org.peerfact.impl.overlay.dht.pastry.base.nodestate.StateHelpers;
import org.peerfact.impl.overlay.dht.pastry.base.operations.AbstractPastryOperation;
import org.peerfact.impl.overlay.dht.pastry.base.operations.StoreOperation;
import org.peerfact.impl.overlay.dht.pastry.parallel.messages.IterativeUpdateMsg;
import org.peerfact.impl.overlay.dht.pastry.parallel.operations.LookupOperation;
import org.peerfact.impl.overlay.dht.pastry.parallel.operations.ValueLookupOperation;
import org.peerfact.impl.simengine.Simulator;
import org.peerfact.impl.transport.TransMsgEvent;

/**
 * This class handles incoming messages and dispatches/processes them.
 * 
 * @author Julius Rueckert <peerfact@kom.tu-darmstadt.de>
 * 
 * @version 05/06/2011
 */
public class PastryMessageHandler
		extends
		org.peerfact.impl.overlay.dht.pastry.base.components.PastryMessageHandler {

	/**
	 * true if parallel lookups should happen after every Hop
	 */
	private boolean resplit;

	/**
	 * true if iterative Routing should be used
	 */
	public boolean iterative;

	/**
	 * @param node
	 *            the owning pastry node new constructor by Martin Wiegand.
	 */
	public PastryMessageHandler(PastryNode node,
			boolean resplitt, boolean iterative) {
		super(node);
		this.resplit = resplitt;
		this.iterative = iterative;
	}

	@Override
	public void messageArrived(TransMsgEvent receivingEvent) {
		Message msg = receivingEvent.getPayload();

		if (!(msg instanceof PastryBaseMsg)) {
			return;
		}

		PastryBaseMsg pMsg = (PastryBaseMsg) msg;
		PastryContact senderContact = new PastryContact(pMsg.getSender(),
				receivingEvent.getSenderTransInfo());

		if (pMsg.getSender() != null) {
			node.addContact(new PastryContact(pMsg.getSender(), receivingEvent
					.getSenderTransInfo()));
		}

		// TODO check if acknowledgment can be combined with possible answer
		/*
		 * Instead of a plain AckMsg each Block may decide to overwrite ackMsg
		 * to piggyback the ACK. It is important to note that a Message sent via
		 * reply will not trigger messageArrived() at the receiver but instead
		 * fire the provided callback.
		 */
		Message ackMsg = new AckMsg();

		// implementation for IterativeUpdateMsg handling by Martin Wiegand
		if (msg instanceof IterativeUpdateMsg) {
			IterativeUpdateMsg iMsg = (IterativeUpdateMsg) msg;

			int lookupId = iMsg.getOperationID();

			if (((PastryNode) node).containsOperation(lookupId)) {
				// only handle the message if the operation is registered with
				// the node
				if (resplit) {
					// splitting up after every hop can be done but increases
					// messages sent dramatically
					if (iMsg.getValueLookup()) {
						// this branch is entered if the its a valueLookup. This
						// only changes the message Type thati s forwarded.
						RouteSet rs = iMsg.getRouteSet();
						for (int i = 0; i < java.lang.Math.min(
								((PastryNode) node)
										.getParallelLookups(), rs.getSize()); i++) {
							// send lookup messages to as many nodes as possible
							// until node.getParallelLookups() messages have
							// been sent
							PastryContact currentReceiver = StateHelpers
									.getClosestContact(iMsg.getLookupKey(),
											new LinkedList<PastryContact>(), rs);
							rs.removeEntry(currentReceiver);
							// notify monitor about forwarded message
							Simulator.getMonitor().dhtLookupForwarded(
									iMsg.getSenderContact(),
									iMsg.getLookupKey().getCorrespondingKey(),
									currentReceiver, iMsg.getHops());
							ValueLookupMsg reply = new ValueLookupMsg(node
									.getOverlayContact(), currentReceiver
									.getOverlayID(), iMsg.getLookupKey(), iMsg
									.getOperationID(), iMsg.getHops() + 1);
							// forward message
							forwardMsg(reply, currentReceiver.getOverlayID()
									.getCorrespondingKey());
						}
						for (PastryContact c : rs) {
							node.addContact(c);
						}
					} else {

						RouteSet rs = iMsg.getRouteSet();
						for (int i = 0; i < java.lang.Math.min(
								((PastryNode) node)
										.getParallelLookups(), rs.getSize()); i++) {
							// send lookup messages to as many nodes as possible
							// until node.getParallelLookups() messages have
							// been sent
							PastryContact currentReceiver = StateHelpers
									.getClosestContact(iMsg.getLookupKey(),
											new LinkedList<PastryContact>(), rs);
							rs.removeEntry(currentReceiver);
							Simulator.getMonitor().dhtLookupForwarded(
									iMsg.getSenderContact(),
									iMsg.getLookupKey().getCorrespondingKey(),
									currentReceiver, iMsg.getHops());
							LookupMsg reply = new LookupMsg(node
									.getOverlayContact(), currentReceiver, iMsg
									.getLookupKey(), iMsg.getOperationID(),
									iMsg.getHops() + 1);
							// forward message
							forwardMsg(reply, currentReceiver.getOverlayID()
									.getCorrespondingKey());
						}
						for (PastryContact c : rs) {
							node.addContact(c);
						}

					}
				} else {
					PastryContact currentReceiver;
					// get the current receiver.
					if (iMsg.getRouteSet().getSize() > 1) {
						currentReceiver = StateHelpers.getClosestContact(iMsg
								.getLookupKey(),
								new LinkedList<PastryContact>(), iMsg
										.getRouteSet());
					} else {
						currentReceiver = iMsg.getRouteSet().iterator().next();
					}

					// System.out.print(" OP-ID:" + iMsg.getOperationID()
					// + " Target:" + iMsg.getLookupKey() + "Best: "
					// + currentReceiver.getOverlayID() + " contains: ");
					for (PastryContact c : iMsg.getRouteSet()) {
						System.out.print(c.getOverlayID() + ", ");
					}
					System.out.println();
					node.addContact(currentReceiver);

					// forward the specific message to it.
					if (iMsg.getValueLookup()) {
						Simulator.getMonitor().dhtLookupForwarded(
								iMsg.getSenderContact(),
								iMsg.getLookupKey().getCorrespondingKey(),
								currentReceiver, iMsg.getHops());
						ValueLookupMsg reply = new ValueLookupMsg(node
								.getOverlayContact(), currentReceiver
								.getOverlayID(), iMsg.getLookupKey(), iMsg
								.getOperationID(), iMsg.getHops() + 1);
						forwardMsg(reply, currentReceiver.getOverlayID()
								.getCorrespondingKey());
					} else {
						// System.out.println(" OP-ID:" + iMsg.getOperationID()
						// + " Sender:" + node.getOverlayID()
						// + " Receiver:" + currentReceiver.getOverlayID()
						// + " Target: " + iMsg.getLookupKey()
						// + " hop count:" + iMsg.getHops()
						// + " new Lookup message.");
						Simulator.getMonitor().dhtLookupForwarded(
								iMsg.getSenderContact(),
								iMsg.getLookupKey().getCorrespondingKey(),
								currentReceiver, iMsg.getHops());
						LookupMsg reply = new LookupMsg(node
								.getOverlayContact(), currentReceiver, iMsg
								.getLookupKey(), iMsg.getOperationID(), iMsg
								.getHops());
						reply.incrementHopCount();
						forwardMsg(reply, currentReceiver.getOverlayID()
								.getCorrespondingKey());
					}
				}
			}
		}

		else if (msg instanceof JoinMsg) {
			JoinMsg jMsg = (JoinMsg) msg;
			PastryKey key = jMsg.getKey();

			StateUpdateMsg reply;

			if (node.isResponsibleFor(key)) {
				// Mark the reply as final reply
				reply = new FinalJoinAnswerMsg(node.getOverlayID(), jMsg
						.getSender(), node.getAllNeighbors());
			} else {
				reply = new StateUpdateMsg(node.getOverlayID(), jMsg
						.getSender(), node.getAllNeighbors());

				// Forward the JoinMessage
				forwardMsg(msg, key);
			}
			MsgTransInfo<PastryContact> replyInfo = new MsgTransInfo<PastryContact>(
					reply, jMsg.getSenderContact());
			sendMsg(replyInfo);

		} else if (msg instanceof StateUpdateMsg) {
			StateUpdateMsg rMsg = (StateUpdateMsg) msg;
			Collection<PastryContact> newContacts = rMsg.getContacts();

			long leafSetLastChanged = rMsg.getLeafSetTimestamp();
			if (leafSetLastChanged != -1
					&& leafSetLastChanged < node.getLeafSet().getLastChanged()) {
				/*
				 * Send back a more recent leaf set
				 */

				StateUpdateMsg informMsg = new StateUpdateMsg(node
						.getOverlayID(), pMsg.getSender(), node
						.getLeafSetNodes());

				MsgTransInfo<PastryContact> replyInfo = new MsgTransInfo<PastryContact>(
						informMsg, senderContact);
				sendMsg(replyInfo);
			}

			// Insert new contacts into state tables
			node.addContacts(newContacts);

			// If this was the final answer, finish join operation
			if (rMsg instanceof FinalJoinAnswerMsg && unfinishedJoinOp != null) {

				Collection<PastryContact> allNeighbors = node.getAllNeighbors();
				StateUpdateMsg informMsg;

				for (PastryContact n : allNeighbors) {
					informMsg = new StateUpdateMsg(node.getOverlayID(), n
							.getOverlayID(), allNeighbors, node.getLeafSet()
							.getLastChanged());

					MsgTransInfo<PastryContact> replyInfo = new MsgTransInfo<PastryContact>(
							informMsg, n);
					sendMsg(replyInfo);
				}

				// Inform about finished join
				unfinishedJoinOp.joinOperationFinished();
				unfinishedJoinOp = null;
			}
		} else if (msg instanceof RequestLeafSetMsg) {
			/*
			 * Answer the request with this node's leafset entries
			 */
			RequestLeafSetMsg rMsg = (RequestLeafSetMsg) msg;
			Collection<PastryContact> leafSetNodes = node.getLeafSetNodes();
			StateUpdateMsg reply = new StateUpdateMsg(node.getOverlayID(), rMsg
					.getSender(), leafSetNodes, node.getLeafSet()
					.getLastChanged());

			sendMsg(new MsgTransInfo<PastryContact>(reply, senderContact));

		} else if (msg instanceof RequestNeighborhooodSetMsg) {
			/*
			 * Answer the request with this node's neighborhoodset entries
			 */
			RequestNeighborhooodSetMsg rMsg = (RequestNeighborhooodSetMsg) msg;
			Collection<PastryContact> neighbors = node.getNeighborSetNodes();
			StateUpdateMsg reply = new StateUpdateMsg(node.getOverlayID(), rMsg
					.getSender(), neighbors);

			sendMsg(new MsgTransInfo<PastryContact>(reply, senderContact));

		} else if (msg instanceof RequestRouteSetMsg) {
			/*
			 * Answer the request with this node's entries for given
			 * row/column-pair in Routing Table
			 */
			RequestRouteSetMsg rMsg = (RequestRouteSetMsg) msg;
			PastryRoutingTable routingTable = node
					.getPastryRoutingTable();
			Collection<PastryContact> neighbors = new LinkedHashSet<PastryContact>();
			RouteSet entries = routingTable.getEntrySet(rMsg.getRow(), rMsg
					.getCol());
			if (entries != null) {
				for (PastryContact c : entries) {
					neighbors.add(c);
				}
			}
			// overwrite ACK to include the StateUpdate
			ackMsg = new StateUpdateMsg(rMsg.getReceiver(), rMsg.getSender(),
					neighbors);
		} else if (msg instanceof LookupMsg) {
			LookupMsg lMsg = (LookupMsg) msg;

			if (node.isResponsibleFor(lMsg.getTarget().getCorrespondingKey())) {
				// Send a reply
				node.addContact(lMsg.getSenderContact());
				LookupReplyMsg reply = new LookupReplyMsg(node
						.getOverlayContact(), lMsg.getSenderContact(), lMsg);

				MsgTransInfo<PastryContact> replyInfo = new MsgTransInfo<PastryContact>(
						reply, lMsg.getSenderContact());

				sendMsg(replyInfo);
			} else { // modified by Martin Wiegand.
				// only forward messages with less than MSG_MAX_HOPS hops
				// (hack to avoid too high hop counts)
				if (lMsg.getHops() < PastryConstants.MSG_MAX_HOPS) {
					// check if iterative lookups are used
					if (!iterative) {
						// Forward the request
						// Notify Monitor
						if (resplit) {
							// if lookups are split up again, when using
							// parallel lookups
							node.addContact(lMsg.getSenderContact());
							lMsg.incrementHopCount();

							RouteSet rs = ((PastryNode) node).getNextHops(lMsg
									.getTarget());
							Simulator.getMonitor().dhtLookupForwarded(
									lMsg.getSenderContact(),
									lMsg.getTarget().getCorrespondingKey(),
									node.getOverlayContact(), lMsg.getHops());
							// notify monitor that lookup was forwarded.

							for (int i = 0; i < java.lang.Math
									.min(((PastryNode) node)
											.getParallelLookups(), rs.getSize()); i++) {
								PastryContact currentReceiver = StateHelpers
										.getClosestContact(
												lMsg.getTarget(),
												new LinkedList<PastryContact>(),
												rs);
								rs.removeEntry(currentReceiver);
								// forward lookup messages
								forwardMsg(lMsg, currentReceiver.getOverlayID()
										.getCorrespondingKey());
							}
						} else {
							// original code.
							node.addContact(lMsg.getSenderContact());

							lMsg.incrementHopCount();

							Simulator.getMonitor().dhtLookupForwarded(
									lMsg.getSenderContact(),
									lMsg.getTarget().getCorrespondingKey(),
									node.getOverlayContact(), lMsg.getHops());

							forwardMsg(lMsg, lMsg.getTarget()
									.getCorrespondingKey());
						}
					} else {
						node.addContact(lMsg.getSenderContact());
						RouteSet rs = null;

						if (!resplit) {
							rs = new RouteSet();
							rs.insertEntry(node.getNextHop(lMsg.getTarget()));
							System.out.println("None resplit.");
						} else {
							rs = ((PastryNode) node).getNextHops(lMsg
									.getTarget());
						}

						lMsg.incrementHopCount();
						Simulator.getMonitor().dhtLookupForwarded(
								lMsg.getSenderContact(),
								lMsg.getTarget().getCorrespondingKey(),
								node.getOverlayContact(), lMsg.getHops());
						IterativeUpdateMsg reply = new IterativeUpdateMsg(node
								.getOverlayContact(), lMsg.getSenderContact(),
								lMsg.getTarget(), rs, lMsg.getLookupId(), lMsg
										.getHops(), false);

						node.addContact(lMsg.getSenderContact());

						System.out.println(" OP-ID:" + lMsg.getLookupId()
								+ " Sender:" + node.getOverlayID()
								+ " Receiver:"
								+ lMsg.getSenderContact().getOverlayID()
								+ " iterative Update sent.");

						// forward message are not iterativeUpdateMessages.
						forwardMsg(reply, lMsg.getSenderContact()
								.getOverlayID().getCorrespondingKey());
					}
				}

			}
		} else if (msg instanceof LookupReplyMsg) {
			LookupReplyMsg lMsg = (LookupReplyMsg) msg;

			// Get the operation that requested the lookup
			int lookupId = lMsg.getRequest().getLookupId();
			AbstractPastryOperation<?> op = node.unregisterOperation(lookupId);

			// Deliver the result if the operation still exists
			if (op != null && op instanceof LookupOperation) {
				LookupOperation lookupOp = (LookupOperation) op;
				if (lookupOp.getResult().isEmpty()) {
					lookupOp.deliverResult(lMsg.getResponsibleContact(), lMsg
							.getRequest().getTarget(), lMsg.getRequest()
							.getLookupId(), lMsg.getRequest().getHops());
					// System.out.println(" OP-ID:"
					// + lMsg.getRequest().getLookupId() + " Initiator:"
					// + node.getOverlayID() + " Responsible Node:"
					// + lMsg.getSender() + " Lookup Reply. Hops:"
					// + lMsg.getRequest().getHops());
				}
			} else {
				log
						.debug("lookup operation not found -> duplicated answer to lookup - receiver = "
								+ node.getOverlayID()
								+ " lookupId = "
								+ lookupId);
			}

		} else if (msg instanceof StoreMsg) {
			StoreMsg storeMsg = (StoreMsg) msg;
			if (node.isResponsibleFor(storeMsg.getKey().getCorrespondingKey())) {
				// node is responsible to store the DHTObject with the
				// respective key
				log.debug("PastryNode " + node.getOverlayID()
						+ " is maintainer of the DHTObject with key "
						+ storeMsg.getKey() + " from node "
						+ storeMsg.getSender());
				node.getDHT().addDHTEntry(
						storeMsg.getKey().getCorrespondingKey(),
						storeMsg.getValue());
				// create answer for the storing peer
				Set<PastryContact> storingPeers = new LinkedHashSet<PastryContact>();
				storingPeers.add(node.getOverlayContact());
				ConfirmStoreMsg confirmMsg = new ConfirmStoreMsg(node
						.getOverlayContact(), storeMsg.getSenderContact(),
						storingPeers, storeMsg.getOperationID());
				sendMsg(new MsgTransInfo<PastryContact>(confirmMsg, storeMsg
						.getSenderContact()));
			} else {
				// node is not responsible for the DHTObject and has to forward
				// it
				storeMsg.incrementHops();
				forwardMsg(storeMsg, storeMsg.getKey().getCorrespondingKey());
			}
		} else if (msg instanceof ConfirmStoreMsg) {
			ConfirmStoreMsg csMsg = (ConfirmStoreMsg) msg;
			AbstractPastryOperation<?> op = node.unregisterOperation(Integer
					.valueOf(csMsg.getOperationID()));
			if (op != null) {
				((StoreOperation) op).deliverResult(csMsg);
			}
		}

		// handle the value-lookup-related messages
		else if (msg instanceof ValueLookupMsg) {
			ValueLookupMsg lookupMsg = (ValueLookupMsg) msg;
			if (node.isResponsibleFor(lookupMsg.getKey().getCorrespondingKey())) {
				// node is responsible to answer with the corresponding
				// DHTObject for the respective key

				DHTObject value = (DHTObject) node.getDHT().getDHTValue(
						lookupMsg.getKey().getCorrespondingKey());
				if (value == null) {
					log
							.warn("PastryNode "
									+ node.getOverlayID()
									+ " is maintainer but does not have the DHTObject with key "
									+ lookupMsg.getKey()
									+ " for requesting node "
									+ lookupMsg.getSender());
				} else {
					log.info("PastryNode " + node.getOverlayID()
							+ " is maintainer of the DHTObject with key "
							+ lookupMsg.getKey() + " for requesting node "
							+ lookupMsg.getSender());
				}
				// create answer for the requesting peer
				ValueLookupReplyMsg replyMsg = new ValueLookupReplyMsg(node
						.getOverlayContact(), lookupMsg.getSender(), value,
						lookupMsg.getOperationID(), lookupMsg.getHops());
				sendMsg(new MsgTransInfo<PastryContact>(replyMsg, lookupMsg
						.getSenderContact()));
			} else {
				// node is not responsible for the DHTObject and has to forward
				// it
				lookupMsg.incrementHops();
				forwardMsg(lookupMsg, lookupMsg.getKey().getCorrespondingKey());
			}
		} else if (msg instanceof ValueLookupReplyMsg) {
			ValueLookupReplyMsg replyMsg = (ValueLookupReplyMsg) msg;
			AbstractPastryOperation<?> op = node.unregisterOperation(Integer
					.valueOf(replyMsg.getOperationID()));
			if (op != null) {
				((ValueLookupOperation) op).deliverResult(replyMsg);
			}
		}

		// TODO Handle other message types if needed

		/*
		 * ACK for the message. This Message may be overwritten by a Messages
		 * ElseIf-Block
		 */
		sendReply(receivingEvent, ackMsg);
	}

	protected int forwardMsg(Message msg, final PastryContact c) {
		/*
		 * Forward message
		 */
		final PastryContact nextHop = c;
		MsgTransInfo<PastryContact> toSend = new MsgTransInfo<PastryContact>(
				msg, nextHop);

		Failed transFailedCallback = new Failed() {

			private int retries = 0;

			@Override
			public void transmissionFailed(Message message) {

				// Remove failed hop
				node.removeNeighbor(nextHop);

				if (retries > PastryConstants.MSG_MAX_ALTERNATIVE_HOPS) {
					return;
				}

				// Get an alternative
				PastryContact nextHop1 = node.getNextHop(c.getOverlayID());

				if (nextHop1 != null && nextHop1 != node.getOverlayContact()) {
					MsgTransInfo<PastryContact> toSend1 = new MsgTransInfo<PastryContact>(
							message, nextHop1);
					retries++;

					sendMsg(toSend1, this);
				}
			}
		};

		return sendMsg(toSend, transFailedCallback);
	}

}
